package clusteringTest;




import java.io.BufferedWriter;
import java.io.File;
import java.io.FileWriter;
import java.net.URI;
import java.text.DateFormat;
import java.text.SimpleDateFormat;
import java.util.ArrayList;
import java.util.Collections;
import java.util.Comparator;
import java.util.Date;
import java.util.HashMap;
import java.util.HashSet;
import java.util.Iterator;
import java.util.LinkedList;
import java.util.List;
import java.util.Map;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.SequenceFile;
import org.apache.mahout.clustering.WeightedVectorWritable;
import org.apache.mahout.clustering.canopy.CanopyDriver;
import org.apache.mahout.clustering.kmeans.Cluster;
import org.apache.mahout.clustering.kmeans.KMeansDriver;
import org.apache.mahout.common.distance.CosineDistanceMeasure;
import org.apache.mahout.common.distance.EuclideanDistanceMeasure;
import org.apache.mahout.common.distance.TanimotoDistanceMeasure;

import test.ReaderFromHDFS;



public class KMeansClustering {
	
	final static String CRLN = System.getProperty("line.separator");
	  @SuppressWarnings("unchecked")
	public void ClusteringTexts(String page_title, String page_date,String clustering_result) throws Exception {
//		  String page_title = "blog_page";//
//		  String page_date = "2010-12-08";//
//		  String page_title = args[0];//
//		  String page_date = args[1];
		  
		  
		HbaseToVector hv= new HbaseToVector();
		hv.TransformToVector(page_title,page_date);
		
		CreateMapFromDic  cmd  = new CreateMapFromDic();	
		HashMap<String,String> map = cmd.createMapFromDic("hdfs://localhost:9000/user/tuxinhui/clustering/dic");
        //判断如果habase中没有数据，则退出处理
		if(map.size()==0)
		{
			System.exit(0);
		}
		
	    Configuration conf = new Configuration();
	    FileSystem fs = FileSystem.get(conf);
	    
	    
	    Path vectorsFolder = new Path("hdfs://localhost:9000/user/tuxinhui/clustering/points/file1");
	    
	    
	    Path clusterOutput = new Path("hdfs://localhost:9000/user/tuxinhui/clustering/output");
	    Path canopyCentroids = new Path(clusterOutput,"canopy-centroids");    

	    
	    CanopyDriver.run(vectorsFolder, canopyCentroids,
	    		new CosineDistanceMeasure(), 0.1, 0.1,false, false);
	    
	    KMeansDriver.run(conf, vectorsFolder,new Path(canopyCentroids,"clusters-0"), 
	    		clusterOutput, new CosineDistanceMeasure(), 0.01,10, true, false);
	    
	    String uri = "hdfs://localhost:9000/user/tuxinhui/clustering/output/clusteredPoints/";
	    fs = FileSystem.get(URI.create(uri),conf);
	    SequenceFile.Reader reader = new SequenceFile.Reader(fs,
	        new Path("hdfs://localhost:9000/user/tuxinhui/clustering/output/" + Cluster.CLUSTERED_POINTS_DIR
	                 + "/part-m-00000"), conf);
	    
	    IntWritable key = new IntWritable();
	    WeightedVectorWritable value = new WeightedVectorWritable();
	    /*while (reader.next(key, value)) {
	      System.out.println(value.toString() + " belongs to cluster "
	                         + key.toString());
	    }*/
	    String resultrecord = null;
	    int leftparentheses = 0;
	    int rightparentheses = 0;
	    String vectorstring = null;
	    
	    HashMap<String, Integer> cmap = new HashMap <String, Integer>();//camp用来 存放聚类类别，及每个类别中包含的文档数量
	    while (reader.next (key, value))
	    {
	    	resultrecord = value.toString();
	    	leftparentheses = resultrecord.indexOf("[");
	    	rightparentheses = resultrecord.indexOf("]");
	    	vectorstring = resultrecord.substring(leftparentheses+1, rightparentheses);
	    	
	    	System.out.println(vectorstring);
	    	
	    	System.out.println(map.get(vectorstring) + " belongs to cluster " + key.toString());
	    	//统计有多少个类别，以及每个类别中有多少篇文档
	    	//HashMap<String, Integer> cmap = new HashMap <String, Integer>();//camp用来 存放聚类类别，及每个类别中包含的文档数量
	    	int count = 0;
	    	String ckey = key.toString();//ckey:类名序号
	    	
	    	if (cmap.get(ckey)!=null)
			{
				count = cmap.get(ckey) + 1;
				cmap.remove(ckey);
				cmap.put(ckey, count);
			}
			else
			{
				cmap.put(ckey, 1);
			}
	    }//while
	    //显示统计结束后的结果
	    for (Iterator iter = cmap.keySet().iterator();iter.hasNext();)
		{
	    	//objectkey:文档类别(cluster)号
	    	//objectvalue:类别(cluster)中包含的文档数量
			Object objectkey = iter.next();
			Object objectvalue = cmap.get(objectkey);
			System.out.printf ("cluster:" + objectkey + " ");
			System.out.println("has:"+objectvalue +"documents");
		}
	    //sort, 按照文档的数量降序排列类别
	    @SuppressWarnings({"unchecked", "rawtypes" })
	    //把map中的<key,value>对导入到链表中，然后按照value对链表排序
		List sorted_map = new LinkedList(cmap.entrySet());

		Collections.sort (sorted_map, 
				new Comparator() {
          public int compare(Object o1, Object o2) {
               return ((Comparable) ((Map.Entry) (o2)).getValue())
              .compareTo(((Map.Entry) (o1)).getValue());
          }//compare()
		}//Comparator
		);//Collection.sort()
		
		//遍历链表中的<key,value>对
		ArrayList<String> arrayListOfHashMapItem = new ArrayList<String>(); 
		Iterator iterator = sorted_map.iterator();
		int flag = 1;
		while (iterator.hasNext())
		{
			Map.Entry entry = (Map.Entry)iterator.next();
			//listmapkey:list中map元素的key
			//listmapvalue:list中map元素的value
			Object listmapkey = entry.getKey();
			Object listmapvalue =entry.getValue();
			System.out.println("cluster" + listmapkey.toString() + " : has " + listmapvalue + " docs");
			arrayListOfHashMapItem.add(listmapkey.toString());
			if (flag++ == 10)//只取前10条记录
				break;
		}//while get hs
//		System.out.println(hs.size());
		//查询该类别（cluster）包含文档的title，并将网址，title写入文件
		reader.close();
		
		
	    SequenceFile.Reader readerfromsequence = new SequenceFile.Reader(fs,
		        new Path("hdfs://localhost:9000/user/tuxinhui/clustering/output/" + Cluster.CLUSTERED_POINTS_DIR
		                 + "/part-m-00000"), conf);
		
    	LookUpTitleFromURL lt = new LookUpTitleFromURL();
		ArrayList<StringBuffer> arrayListOfStringBuffer = new ArrayList<StringBuffer>();
		ArrayList<Integer> clusterNum = new ArrayList<Integer>();
		String filerecord = "";//存放每一条要写入文件里面的记录，格式：title url
		while (readerfromsequence.next (key, value))
		{
			//判断读出来的结果是否属于这个类
			if(arrayListOfHashMapItem.contains(key.toString()))
			{
				int index = Integer.parseInt(key.toString());
				if(!clusterNum.contains(index))
				{
					arrayListOfStringBuffer.add(index,new StringBuffer());
					clusterNum.add(index);
				}				
				resultrecord = value.toString();
		    	leftparentheses = resultrecord.indexOf("[");
		    	rightparentheses = resultrecord.indexOf("]");
		    	vectorstring = resultrecord.substring(leftparentheses+1, rightparentheses);
		    	System.out.println(page_title+"   "+map.get(vectorstring));
				String title = lt.QueryTitleByURL(page_title, map.get(vectorstring));
				filerecord = title + " " + map.get(vectorstring);			
				System.out.println("网页名称（title）与网页地址（url）分别为： "+filerecord);
				arrayListOfStringBuffer.get(index).append(filerecord);
				arrayListOfStringBuffer.get(index).append(CRLN);
//				StringBuffer stringBuffer = new StringBuffer();
//				stringBuffer.append(filerecord);
			}   	
		}//while  
		readerfromsequence.close();
		
//      不写到HDFS，直接写到本地，方便读取些～
//		System.out.println("开始将文件写入HDFS... ... ...");
//		ReaderFromHDFS RF = new ReaderFromHDFS();
		  
//		Date date = new Date();
//		DateFormat dateFormat = new SimpleDateFormat ("EE-MM-dd-yyyy");
		System.out.println("开始将文件写入本地... ... ...");
		
		File resultfolder = new File(clustering_result);
		if(!resultfolder.exists())
		{
			resultfolder.mkdirs();
		}
		String placetowrite = clustering_result+"/"+page_date;//home/tuxinhui/workspace/analyzer/
		String wholecontent = "";
		BufferedWriter bufferedWriter = new BufferedWriter(new FileWriter(placetowrite));
//		FileWriter fileWriter = new FileWriter(placetowrite);
		for(String cluster:arrayListOfHashMapItem)
		{
			System.out.println("将被写入文件的内容：");
			System.out.println("类别：" + cluster);
			String content = arrayListOfStringBuffer.get(Integer.parseInt(cluster)).toString();
			System.out.println(content);
			wholecontent += content;
			//每个类别的文件间空一行，方便区分
			wholecontent += CRLN;
			
			//RF.WriterHDFS(content,placetowrite);
		}
//		RF.WriterHDFS(wholecontent, placetowrite);
		bufferedWriter.write(wholecontent);
		bufferedWriter.flush();   
		bufferedWriter.close(); 
		System.out.println("恭喜! 已将聚类结果成功写入文件!");
	    
	   fs.deleteOnExit(clusterOutput);	
	    
    }
    public static void main (String[] args)
    {
    	KMeansClustering kmeansClustering = new KMeansClustering();
    	String page_title = "blog_page";//网页名称
		String page_date = "2010-12-08";//网页日期
		String clustering_result = "/home/tuxinhui/workspace/analyzer/AnalyzerResluts/clusteredresult";//存放结果的文件夹路径
    	try 
    	{
			kmeansClustering.ClusteringTexts(page_title, page_date,clustering_result);
		} 
    	catch (Exception e) 
    	{
			// TODO Auto-generated catch block
			e.printStackTrace();
		}
    }
}